// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#pragma once
#include <nebula/version.h>
#ifdef NEBULA_HDFS
#include <cstdint>
#include <memory>
#include <string>
#include <unordered_map>
#include <vector>

#include <nebula/io/interfaces.h>
#include <turbo/base/macros.h>

namespace nebula {

    class Buffer;

    class MemoryPool;


    namespace io {

        class HdfsReadableFile;

        class HdfsOutputStream;

        /// DEPRECATED.  Use the FileSystem API in nebula::fs instead.
        struct ObjectType {
            enum type {
                FILE, DIRECTORY
            };
        };

        /// DEPRECATED.  Use the FileSystem API in nebula::fs instead.
        struct TURBO_EXPORT FileStatistics {
            /// Size of file, -1 if finding length is unsupported
            int64_t size;
            ObjectType::type kind;
        };

        class TURBO_EXPORT FileSystem {
        public:
            virtual ~FileSystem() = default;

            virtual turbo::Status MakeDirectory(const std::string &path) = 0;

            virtual turbo::Status DeleteDirectory(const std::string &path) = 0;

            virtual turbo::Status GetChildren(const std::string &path,
                                       std::vector<std::string> *listing) = 0;

            virtual turbo::Status Rename(const std::string &src, const std::string &dst) = 0;

            virtual turbo::Status Stat(const std::string &path, FileStatistics *stat) = 0;
        };

        struct HdfsPathInfo {
            ObjectType::type kind;

            std::string name;
            std::string owner;
            std::string group;

            // Access times in UNIX timestamps (seconds)
            int64_t size;
            int64_t block_size;

            int32_t last_modified_time;
            int32_t last_access_time;

            int16_t replication;
            int16_t permissions;
        };

        struct HdfsConnectionConfig {
            std::string host;
            int port;
            std::string user;
            std::string kerb_ticket;
            std::unordered_map<std::string, std::string> extra_conf;
        };

        class TURBO_EXPORT HadoopFileSystem : public FileSystem {
        public:
            ~HadoopFileSystem() override;

            // Connect to an HDFS cluster given a configuration
            //
            // @param config (in): configuration for connecting
            // @param fs (out): the created client
            // @returns turbo::Status
            static turbo::Status Connect(const HdfsConnectionConfig *config,
                                  std::shared_ptr<HadoopFileSystem> *fs);

            // create directory and all parents
            //
            // @param path (in): absolute HDFS path
            // @returns turbo::Status
            turbo::Status MakeDirectory(const std::string &path) override;

            // Delete file or directory
            // @param path absolute path to data
            // @param recursive if path is a directory, delete contents as well
            // @returns error status on failure
            turbo::Status Delete(const std::string &path, bool recursive = false);

            turbo::Status DeleteDirectory(const std::string &path) override;

            // Disconnect from cluster
            //
            // @returns turbo::Status
            turbo::Status Disconnect();

            // @param path (in): absolute HDFS path
            // @returns bool, true if the path exists, false if not (or on error)
            bool Exists(const std::string &path);

            // @param path (in): absolute HDFS path
            // @param info (out)
            // @returns turbo::Status
            turbo::Status GetPathInfo(const std::string &path, HdfsPathInfo *info);

            // @param nbytes (out): total capacity of the filesystem
            // @returns turbo::Status
            turbo::Status get_capacity(int64_t *nbytes);

            // @param nbytes (out): total bytes used of the filesystem
            // @returns turbo::Status
            turbo::Status GetUsed(int64_t *nbytes);

            turbo::Status GetChildren(const std::string &path, std::vector<std::string> *listing) override;

            /// List directory contents
            ///
            /// If path is a relative path, returned values will be absolute paths or URIs
            /// starting from the current working directory.
            turbo::Status ListDirectory(const std::string &path, std::vector<HdfsPathInfo> *listing);

            /// Return the filesystem's current working directory.
            ///
            /// The working directory is the base path for all relative paths given to
            /// other APIs.
            /// NOTE: this actually returns a URI.
            turbo::Status GetWorkingDirectory(std::string *out);

            /// Change
            ///
            /// @param path file path to change
            /// @param owner pass null for no change
            /// @param group pass null for no change
            turbo::Status Chown(const std::string &path, const char *owner, const char *group);

            /// Change path permissions
            ///
            /// \param path Absolute path in file system
            /// \param mode Mode bitset
            /// \return turbo::Status
            turbo::Status Chmod(const std::string &path, int mode);

            // Move file or directory from source path to destination path within the
            // current filesystem
            turbo::Status Rename(const std::string &src, const std::string &dst) override;

            turbo::Status copy(const std::string &src, const std::string &dst);

            turbo::Status Move(const std::string &src, const std::string &dst);

            turbo::Status Stat(const std::string &path, FileStatistics *stat) override;

            // TODO(wesm): GetWorkingDirectory, SetWorkingDirectory

            // open an HDFS file in READ mode. Returns error
            // status if the file is not found.
            //
            // @param path complete file path
            turbo::Status OpenReadable(const std::string &path, int32_t buffer_size,
                                std::shared_ptr<HdfsReadableFile> *file);

            turbo::Status OpenReadable(const std::string &path, int32_t buffer_size,
                                const io::IOContext &io_context,
                                std::shared_ptr<HdfsReadableFile> *file);

            turbo::Status OpenReadable(const std::string &path, std::shared_ptr<HdfsReadableFile> *file);

            turbo::Status OpenReadable(const std::string &path, const io::IOContext &io_context,
                                std::shared_ptr<HdfsReadableFile> *file);

            // FileMode::WRITE options
            // @param path complete file path
            // @param buffer_size 0 by default
            // @param replication 0 by default
            // @param default_block_size 0 by default
            turbo::Status OpenWritable(const std::string &path, bool append, int32_t buffer_size,
                                int16_t replication, int64_t default_block_size,
                                std::shared_ptr<HdfsOutputStream> *file);

            turbo::Status OpenWritable(const std::string &path, bool append,
                                std::shared_ptr<HdfsOutputStream> *file);

        private:
            friend class HdfsReadableFile;

            friend class HdfsOutputStream;

            class TURBO_NO_EXPORT HadoopFileSystemImpl;

            std::unique_ptr<HadoopFileSystemImpl> impl_;

            HadoopFileSystem();
            TURBO_DISALLOW_COPY_AND_ASSIGN(HadoopFileSystem);
        };

        class TURBO_EXPORT HdfsReadableFile : public RandomAccessFile {
        public:
            ~HdfsReadableFile() override;

            turbo::Status close() override;

            bool closed() const override;

            // NOTE: If you wish to read a particular range of a file in a multithreaded
            // context, you may prefer to use read_at to avoid locking issues
            turbo::Result<int64_t> read(int64_t nbytes, void *out) override;

            turbo::Result<std::shared_ptr<Buffer>> read(int64_t nbytes) override;

            turbo::Result<int64_t> read_at(int64_t position, int64_t nbytes, void *out) override;

            turbo::Result<std::shared_ptr<Buffer>> read_at(int64_t position, int64_t nbytes) override;

            turbo::Status Seek(int64_t position) override;

            turbo::Result<int64_t> tell() const override;

            turbo::Result<int64_t> get_size() override;

        private:
            explicit HdfsReadableFile(const io::IOContext &);

            class TURBO_NO_EXPORT HdfsReadableFileImpl;

            std::unique_ptr<HdfsReadableFileImpl> impl_;

            friend class HadoopFileSystem::HadoopFileSystemImpl;

            TURBO_DISALLOW_COPY_AND_ASSIGN(HdfsReadableFile);
        };

        // Naming this file OutputStream because it does not support seeking (like the
        // WritableFile interface)
        class TURBO_EXPORT HdfsOutputStream : public OutputStream {
        public:
            ~HdfsOutputStream() override;

            turbo::Status close() override;

            bool closed() const override;

            using OutputStream::write;

            turbo::Status write(const void *buffer, int64_t nbytes) override;

            turbo::Status flush() override;

            turbo::Result<int64_t> tell() const override;

        private:
            class TURBO_NO_EXPORT HdfsOutputStreamImpl;

            std::unique_ptr<HdfsOutputStreamImpl> impl_;

            friend class HadoopFileSystem::HadoopFileSystemImpl;

            HdfsOutputStream();

            TURBO_DISALLOW_COPY_AND_ASSIGN(HdfsOutputStream);
        };

        TURBO_EXPORT turbo::Status HaveLibHdfs();

    }  // namespace io
}  // namespace nebula
#endif